package tools

import (
   "fmt"

   "github.com/Shopify/sarama"

   "huaxrobot/config"   
)


type KafkaClient struct {
	address string;	
}

func NewKafka(address string) *KafkaClient {
	kafka := &KafkaClient{}
	kafka.address = address
	return kafka
}

var KAFKA *KafkaClient

func InitKafka() {
	HOST := config.CONFIG.GetString("kafka.host")
	PORT := config.CONFIG.GetString("kafka.port")
	ADDRESS := HOST + ":" + PORT
	KAFKA = NewKafka(ADDRESS)
}

//消息写入kafka
func (kafka *KafkaClient) Producer (topic string, message string) {  
   config := sarama.NewConfig()
   config.Producer.RequiredAcks = sarama.WaitForAll
   config.Producer.Partitioner = sarama.NewRandomPartitioner
   config.Producer.Return.Successes = true
   //生产者
   client, err := sarama.NewSyncProducer([]string{kafka.address}, config)
   if err != nil {
      fmt.Println("producer close,err:", err)
      return
   }
   defer client.Close()  
	//创建消息
	msg := &sarama.ProducerMessage{}
	msg.Topic = topic
	msg.Value = sarama.StringEncoder(message)
	//发送消息
	pid, offset, err := client.SendMessage(msg)
	if err != nil {
		fmt.Println("send message failed,", err)
		return
	}
	fmt.Printf("pid:%v offset:%v\n,", pid, offset)
}

//收到消息的回调函数
type Onmessage func(string)  

//kafka消息接收
func (kafka *KafkaClient) Consumer(topic string, onmessage Onmessage)  {
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true
	config.Version = sarama.V0_11_0_2
	// consumer
	consumer, err := sarama.NewConsumer([]string{kafka.address}, config)
	if err != nil {
		fmt.Printf("consumer_test create consumer error %s\n", err.Error())
		return
	}
	defer consumer.Close()
	partition_consumer, err := consumer.ConsumePartition(topic, 0, sarama.OffsetNewest)
	if err != nil {
		fmt.Printf("try create partition_consumer error %s\n", err.Error())
		return
	}
	defer partition_consumer.Close()
	for {
		select {
		case msg := <-partition_consumer.Messages():
			fmt.Printf("msg offset: %d, partition: %d, timestamp: %s, value: %s\n",
				msg.Offset, msg.Partition, msg.Timestamp.String(), string(msg.Value))
				onmessage(string(msg.Value))
		case err := <-partition_consumer.Errors():
			fmt.Printf("err :%s\n", err.Error())
		}
	}
}